package com.bancomer.pig.functions;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.List;

import org.apache.pig.EvalFunc;
import org.apache.pig.FuncSpec;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.impl.logicalLayer.schema.Schema;

public class CompareDate extends EvalFunc<Long> {
	public Long exec(Tuple input) throws IOException{
		long result = 0;
		try {
	       	 String startDate = (String) input.get(0);
	       	 String endDate = (String) input.get(1);
	       	 Calendar calendarStartDate=Calendar.getInstance();
	       	 calendarStartDate.set(Integer.valueOf(startDate.split("-")[0]), Integer.valueOf(startDate.split("-")[1]), Integer.valueOf(startDate.split("-")[2]),0,0,0);
	       	 Calendar calendarEndtDate=Calendar.getInstance();
	       	 calendarEndtDate.set(Integer.valueOf(endDate.split("-")[0]), Integer.valueOf(endDate.split("-")[1]), Integer.valueOf(endDate.split("-")[2]),0,0,0);
	       	 result = (long)calendarStartDate.compareTo(calendarEndtDate);
		} catch (ExecException ee) {
	           throw ee;
	    }
		return result;
    }

	public Schema outputSchema(Schema input) {
		return new Schema(new Schema.FieldSchema(getSchemaName(this.getClass().getName().toLowerCase(), input), DataType.LONG));
	}
	
    @Override
    public List<FuncSpec> getArgToFuncMapping() throws FrontendException {
        List<FuncSpec> funcList = new ArrayList<FuncSpec>();
        Schema s = new Schema();
        s.add(new Schema.FieldSchema(null, DataType.LONG));
        s.add(new Schema.FieldSchema(null, DataType.LONG));
        funcList.add(new FuncSpec(this.getClass().getName(), s));
        return funcList;
    }
}
